#include<iostream>
#include<string>
#include<vector>
#include<cassert>
#include<string.h>
#include<time.h>
#include<signal.h>
#include<unistd.h>
#include<sys/types.h>
#include <sys/socket.h>  // 主要头文件
#include <netinet/in.h>  // 用于 sockaddr_in
#include <arpa/inet.h>   // 用于 inet_xxx 函数
#include<fcntl.h>
#include<functional>
#include<sys/epoll.h>
#include<unordered_map>
#include<pthread.h>
#include<thread>
#include<mutex>
#include<sys/eventfd.h>
#include<sys/timerfd.h>
#include<any>
#include<typeinfo>
#include<condition_variable>

//日志宏实现
#define INF 0
#define DEB 1
#define ERR 2
#define DEFAULT_LOG_LEVEL DEB
#define LOG(level,format,...) {\
    if(level>=DEFAULT_LOG_LEVEL)\
    {\
        time_t t=time(nullptr);\
        struct tm* lt=localtime(&t);\
        char ts[32]={0};\
        strftime(ts,31,"%H:%M:%S",lt);\
        fprintf(stdout,"[%p %s %s %d] " format "\n",(void*)pthread_self(),ts,__FILE__,__LINE__,##__VA_ARGS__);\
    }\
    }
#define INF_LOG(format,...) LOG(INF,format,##__VA_ARGS__)
#define DEB_LOG(format,...) LOG(DEB,format,##__VA_ARGS__)
#define ERR_LOG(format,...) LOG(ERR,format,##__VA_ARGS__)

#define BUFFER_DEFAULT_SIZE 1024
class Buffer{
    private:
    std::vector<char> _buffer;
    uint64_t _read_idx;
    uint64_t _write_idx;
    public:
    Buffer():_read_idx(0),_write_idx(0),_buffer(BUFFER_DEFAULT_SIZE) {}
    //1.获取_buffer起始地址
    char* Begin() {return &*_buffer.begin();}
    //2.获取当前写的地址
    char* WritePosition() {return Begin()+_write_idx;}
    //3.获取当前读的地址
    char* ReadPosition() {return Begin()+_read_idx;}
    //4.获取缓冲区末尾空闲空间的大小
    uint64_t TailIdleSize() {return _buffer.size()-_write_idx;}
    //5.获取缓冲区起始空闲空间的大小
    uint64_t HeadIdleSize() {return _read_idx;}
    //6.获取可读空间大小
    uint64_t ReadAbleSize() {return _write_idx-_read_idx;}
    //7.将读偏移向后移动
    void MoveReadOffset(uint64_t len){
        if(len<=0) return;
        assert(len<=ReadAbleSize());
        _read_idx+=len;
    }
    //8.将写偏移向后移动
    void MoveWriteOffset(uint64_t len){
        if(len<=0) return;
        assert(len<=TailIdleSize());
        _write_idx+=len;
    }
    //9.确保可写空间足够
    void EnsureWriteSpace(uint64_t len){
        if(len<=TailIdleSize()) return;
        if(len<=HeadIdleSize()+TailIdleSize()){
            uint64_t rsz=ReadAbleSize();
            std::copy(ReadPosition(),ReadPosition()+rsz,Begin());
            _read_idx=0;
            _write_idx=rsz;
        }else{
            DEB_LOG("RESIZE %ld", _write_idx + len);
            _buffer.resize(_write_idx+len);
        }
    }
    //10.写入数据
    void Write(const void* data,uint64_t len){
        if(len<=0) return;
        EnsureWriteSpace(len);
        const char* d=(const char*)data;
        std::copy(d,d+len,WritePosition());
    }
    void WriteAndPush(const void* data,uint64_t len){
        Write(data,len);
        MoveWriteOffset(len);
    }
    void WriteString(const std::string& str){
        Write(str.c_str(),str.size());
    }
    void WriteStringAndPush(const std::string& str){
        Write(str.c_str(),str.size());
        MoveWriteOffset(str.size());
    }
    void WriteBuffer(Buffer& data){
        Write(data.ReadPosition(),data.ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer& data){
        Write(data.ReadPosition(),data.ReadAbleSize());
        MoveWriteOffset(data.ReadAbleSize());
    }
    //11.读取数据
    void Read(void* buf,uint64_t len){
        assert(len<=ReadAbleSize());
        std::copy(ReadPosition(),ReadPosition()+len,(char*)buf);
    }
    void ReadAndPop(void* buf,uint64_t len){
        Read(buf,len);
        MoveReadOffset(len);
    }
    std::string ReadString(uint64_t len){
        assert(len<=ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0],len);
        return str;
    }
    std::string ReadStringAndPop(uint64_t len){
        std::string str=ReadString(len);
        MoveReadOffset(len);
        return str;
    }
    //12.获取第一行数据
    char* FindCRLF(){
        char* res=(char*)memchr(ReadPosition(),'\n',ReadAbleSize());
        return res;
    }
    std::string GetLine(){
        char* pos=FindCRLF();
        if(pos==nullptr) return "";
        return ReadString(pos-ReadPosition()+1);
    }
    std::string GetLineAndPop(){
        std::string str=GetLine();
        MoveReadOffset(str.size());
        return str;
    }
    //13.清空缓冲区
    void Clear(){
        _write_idx=_read_idx=0;
    }
};

//套接字socket类实现
#define MAX_LISTEN 1024
class NetWork{
    public:
    NetWork(){
        DEB_LOG("SIGPIPE INIT");
        signal(SIGPIPE,SIG_IGN);
    }
};
static NetWork nw;
class Socket{
    private:
    int _sockfd;
    public:
    Socket():_sockfd(-1) {}
    Socket(int fd):_sockfd(fd) {}
    ~Socket() { Close(); }
    void Close(){
        if(_sockfd>=0){
            close(_sockfd);
            _sockfd=-1;   //重置为-1，防止重复关闭
        }
    }
    int Fd() { return _sockfd; }
    bool Create(){
        _sockfd=socket(AF_INET,SOCK_STREAM,0);
        if(_sockfd==-1){
            ERR_LOG("socket error: %s",strerror(errno));
            return false;
        }
        return true;
    }
    bool Bind(const std::string& ip,const uint16_t& port){
        struct sockaddr_in addr;
        memset(&addr,0,sizeof(addr));
        addr.sin_family=AF_INET;
        addr.sin_addr.s_addr=inet_addr(ip.c_str());
        addr.sin_port=htons(port);
        int ret=bind(_sockfd,(struct sockaddr*)&addr,sizeof(addr));
        if(ret==-1){
            ERR_LOG("bind error: %s",strerror(errno));
            return false;
        }
        return true;
    }
    bool Listen(int backlog=MAX_LISTEN){
        int ret=listen(_sockfd,backlog);
        if(ret==-1){
            ERR_LOG("listen error: %s",strerror(errno));
            return false;
        }
        return true;
    }
    bool Connect(const std::string& ip,uint16_t port){
        struct sockaddr_in addr;
        addr.sin_family=AF_INET;
        addr.sin_addr.s_addr=inet_addr(ip.c_str());
        addr.sin_port=htons(port);
        int ret=connect(_sockfd,(struct sockaddr*)&addr,sizeof(addr));
        if(ret==-1){
            ERR_LOG("connect error: %s",strerror(errno));
            return false;
        }
        return true;
    }
    int Accept(std::string* client_ip=nullptr,uint16_t* client_port=nullptr){
        struct sockaddr_in addr;
        socklen_t len=sizeof(addr);
        memset(&addr,0,sizeof(addr));
        int newfd=accept(_sockfd,(struct sockaddr*)&addr,&len);
        if(newfd == -1){
            // 🎯 关键修复：区分临时错误和真正错误
            if(errno == EAGAIN || errno == EWOULDBLOCK) {
                // 非阻塞情况下没有连接，这不是错误
                return -1;  // 但不是致命错误，应该返回-1让调用者继续循环
            } else if(errno == EINTR) {
                // 被信号中断，也不是致命错误
                return -1;
            } else {
                // 真正的错误
                ERR_LOG("accept error: %s", strerror(errno));
                return -1;
            }
        }
        // if(!client_ip){
        //     *client_ip=inet_ntoa(addr.sin_addr);
        // }

        // 🎯 修复：正确获取客户端地址信息
        if(client_ip != nullptr) {
            char ip_str[INET_ADDRSTRLEN];
            const char* result = inet_ntop(AF_INET, &addr.sin_addr, ip_str, sizeof(ip_str));
            if(result != nullptr) {
                *client_ip = std::string(ip_str);
            } else {
                *client_ip = "unknown";
            }
        }
        if(client_port!=nullptr){
            *client_port=ntohs(addr.sin_port);
        }
        return newfd;
    }
    int Recv(void* buf,size_t len,int flag=0){
        ssize_t ret=recv(_sockfd,buf,len,flag);
        if(ret<=0){
            //EAGAIN 当前socket的接收缓冲区中没有数据了，在⾮阻塞的情况下才会有这个错误
            //EINTR 表⽰当前socket的阻塞等待，被信号打断了
            if(errno==EAGAIN || errno==EINTR || errno==EWOULDBLOCK){
                return 0;
            }
            ERR_LOG("recv error: %s",strerror(errno));
            return -1;
        }
        return ret;
    }
    //非阻塞接收
    int NonBlockRecv(void* buf,size_t len){
        return Recv(buf,len,MSG_DONTWAIT);
    }
    int Send(const void* buf,size_t len,int flag=0){
        ssize_t ret=send(_sockfd,buf,len,flag);
        if(ret<0){
            if(errno==EAGAIN || errno==EINTR || errno==EWOULDBLOCK){
                return 0;
            }
            ERR_LOG("send error");
            return -1;
        }
        return ret;
        // std::cout<<"ret: "<<ret<<std::endl;
    }
    //非阻塞发送
    int NonBlockSend(const void* buf,size_t len){
        if(len==0) return 0;
        return Send(buf,len,MSG_DONTWAIT);
    }
    //创建一个服务端连接
    bool CreateServer(uint16_t port,const std::string ip="0.0.0.0",bool block_flag=false){
        if(Create()==false) return false;
        if(block_flag) NonBlock();
        ReuseAddress();
        if(Bind(ip,port)==false) return false;
        if(Listen()==false) return false;
        return true;
    }
    //创建一个客户端连接
    bool CreateClient(const std::string& ip,const uint16_t& port){
        if(Create()==false) return false;
        if(Connect(ip,port)==false) return false;
        return true;
    }
    //设置套接字选项----开启地址端口重用
    void ReuseAddress(){
        int val=1;
        //SO_REUSEADDR--地址重用
        setsockopt(_sockfd,SOL_SOCKET,SO_REUSEADDR,(void*)&val,sizeof(int));
        val=1;
        //SO_REUSEPORT--端口重用
        setsockopt(_sockfd,SOL_SOCKET,SO_REUSEPORT,(void*)&val,sizeof(int));
    }
    //设置套接字阻塞属性---设置为非阻塞
    bool NonBlock(){
        int flag = fcntl(_sockfd, F_GETFL, 0);
        if(flag == -1) {
            ERR_LOG("fcntl F_GETFL error: %s", strerror(errno));
            return false;
        }
        if(fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK) == -1) {
            ERR_LOG("fcntl F_SETFL error: %s", strerror(errno));
            return false;
        }
        return true;
    }
};


class Pool;
class EventLoop;
//channel类——我把它理解为Redactor模型里面的事件处理器的组件——事件处理器：将文件描述符与其读写事件的回调关联起来
class Channel{
    private:
    int _fd;
    // Pool* _loop;
    EventLoop* _loop;
    uint32_t _events;   //当前需要监控的事件
    uint32_t _revents;   //当前连接触发的事件
    using EventCallback=std::function<void()>;
    EventCallback _read_callback;    //可读事件触发的回调函数
    EventCallback _write_callback;   //可写事件触发的回调函数
    EventCallback _error_callbakc;   //错误事件触发的回调函数
    EventCallback _close_callback;   //连接断开事件触发的回调函数
    EventCallback _event_callback;   //任意事件触发的回调函数
    public:
    Channel(EventLoop* loop,int fd):_loop(loop),_fd(fd),_events(0),_revents(0)
    {}
    int Fd() { return _fd; }
    uint32_t Events() { return _events; }
    void SetRevents(uint32_t revents) { _revents=revents; }
    
    void SetReadcallback(const EventCallback& cb) { _read_callback=cb; }
    void SetWritecallback(const EventCallback& cb) { _write_callback=cb; }
    void SetErrorcallback(const EventCallback& cb) { _error_callbakc=cb; }
    void SetClosecallback(const EventCallback& cb) { _close_callback=cb; }
    void SetEventcallback(const EventCallback& cb) { _event_callback=cb; }
    bool ReadAble() { return (_events&EPOLLIN); }     //当前是否监控了可读
    bool WriteAble() { return (_events&EPOLLOUT); }    //当前是否监控了可写
    void EnableRead() { _events |=EPOLLIN; Update(); }    //启动读事件监控
    void EnableWrite() { _events |=EPOLLOUT; Update(); }   //启动写事件监控
    void DisableRead() { _events &=~EPOLLIN; Update(); }   //关闭读事件监控
    void DisableWrite() { _events &=~EPOLLOUT; Update(); }   //关闭写事件监控
    void DisableAll() { _events=0; Update(); }      //关闭所有事件监控
    //移除监控
    void Remove();
    void Update();    //将相关的事件监控的修改同步到 EventLoop 的事件监控中去
    //事件处理，一旦触发了事件就调用这个函数，根据触发事件的类型决定调用什么函数
    void HandleEvent(){
        //因为下面四个事件都属于任意一种事件，所以不管下面四个事件哪个事件触发，都会触发任意事件
        if(_event_callback) _event_callback();
        //上面的说法也不完全正确，因为如果连接被释放了自然就不会调用任意事件，如果没有释放我们就可以调用任意事件
        //通过任意事件的调用，我们可以刷新连接的时间，因为刚操作完，所以时间最新，防止被判定为不活跃连接被挂掉

        if((_revents&EPOLLIN) || (_revents&EPOLLRDHUP) || (_revents&EPOLLPRI)){
            if(_read_callback) _read_callback();
        }
        //有可能会释放连接的操作（读数据、触发错误事件、连接挂断），一次只处理一个
        if(_revents&EPOLLOUT){
            if(_write_callback) _write_callback();
        }else if(_revents&EPOLLERR){
            if(_error_callbakc) _error_callbakc();
        }else if(_revents&EPOLLHUP){
            if(_close_callback) _close_callback();
        }
    }
};


//Pool类——事件监控模块（添加、修改、移除事件监控），其实就是对epoll的封装，初始分发器
#define EPOLLEVENTS 1024
class Pool{
    private:
    int _epfd;
    struct epoll_event _evs[EPOLLEVENTS];
    std::unordered_map<int,Channel*> _channels;
    private:
    void Update(Channel* channel,int op){
        int fd=channel->Fd();
        struct epoll_event event;
        event.data.fd=fd;
        event.events=channel->Events();
        // std::cout<<"epfd: "<<_epfd<<" fd: "<<fd<<" events: "<<event.events<<" op: "<<op<<std::endl;
        int ret=epoll_ctl(_epfd,op,fd,&event);
        if(ret<0){
            ERR_LOG("epoll_ctr error: %s",strerror(errno));
        }
        return;
    }
    bool HasChannel(Channel* channel){
        auto it=_channels.find(channel->Fd());
        if(it==_channels.end()){
            return false;
        }
        return true;
    }
    public:
    Pool() {
        _epfd=epoll_create(EPOLLEVENTS);
        if(_epfd<0){
            ERR_LOG("epoll_create error");
            abort();
        }
    }
    ~Pool(){
        if(_epfd>=0){
            close(_epfd);
        }
    }
    //添加或修改监控事件
    void UpdateEvent(Channel* channel){
        bool ret=HasChannel(channel);
        if(ret==false){
            _channels.insert(std::make_pair(channel->Fd(),channel));
         //   std::cout << "_channels.insert" << channel->Fd() << std::endl;
            return Update(channel,EPOLL_CTL_ADD);
        }
       // std::cout << "chanel->fd" << channel->Fd() << std::endl;
        Update(channel,EPOLL_CTL_MOD);
    }
    //移除监控事件
    void RemoveEvent(Channel* channle){
        auto it=_channels.find(channle->Fd());
        if(it!=_channels.end()){
            _channels.erase(it);
        }
        Update(channle,EPOLL_CTL_DEL);
    }
    //开始监控，返回活跃连接
    void Poll(std::vector<Channel*>* active){
        int nfs=epoll_wait(_epfd,_evs,EPOLLEVENTS,-1);
        if(nfs<0){
            if(errno==EINTR){
                return;
            }
            ERR_LOG("epoll_wait error: %s\n",strerror(errno));
            abort();   //退出程序
        }
        for(int i=0;i<nfs;i++){
            // std::cout<<"active fd: "<<_evs[i].data.fd<<std::endl;
            auto it=_channels.find(_evs[i].data.fd);
            assert(it!=_channels.end());
            it->second->SetRevents(_evs[i].events);
            active->push_back(it->second);
        }
    }
};


//定时任务管理类：时间轮+定时器（timerfd）
using TaskFunc=std::function<void()>;
using ReleaseFunc=std::function<void()>;
class TimerTask{
    private:
    uint64_t _id;    //定时器任务对象id，这个不是系统生成的，是我们自己传入的
    uint32_t _timeout;    //定时任务的超时时间
    bool _canceled;    //定时任务是否取消
    TaskFunc _task_cb;     //定时任务
    ReleaseFunc _release;    //用于删除TimerWheel中的定时器对象
    public:
    TimerTask(uint64_t id,uint32_t timeout,TaskFunc cb)
    :_id(id),_timeout(timeout),_task_cb(cb),_canceled(false)
    {}
    ~TimerTask(){
        if(_canceled==false) _task_cb();
        _release();
    }
    void Cancel() { _canceled=true; }
    void SetRelease(const ReleaseFunc& release) { _release=release; }
    uint32_t Delaytime() { return _timeout; }
};
class TimerWheel{
    private:
    using WeakTask=std::weak_ptr<TimerTask>;
    using PtrTask=std::shared_ptr<TimerTask>;
    int _tick;     //当前指针位置
    int _capacity;    //当前轮子容量
    std::vector<std::vector<PtrTask>> _wheel;
    std::unordered_map<uint64_t,WeakTask> _timers;

    EventLoop* _loop;
    int _timerfd;
    std::unique_ptr<Channel> _timer_channel;
    private:
    void RemoveTimer(uint64_t id){
        auto it=_timers.find(id);
        if(it!=_timers.end()){
            _timers.erase(it);
        }
    }
    static int CreateTimerfd(){
        int timerfd=timerfd_create(CLOCK_MONOTONIC,0);   
        //第一个参数需要选择相对时间还是绝对时间，第二个需要选择阻塞还是非阻塞
        if(timerfd<0){
            ERR_LOG("timerfd_create error: %s\n",strerror(errno));
            abort();
        }
        struct itimerspec itime;
        itime.it_value.tv_sec=1;     //第一次超时时间为1秒后
        itime.it_value.tv_nsec=0;
        itime.it_interval.tv_sec=1;    //第一次超时后下一次超时时间也设置为1秒后
        itime.it_interval.tv_nsec=0;
        timerfd_settime(timerfd,0,&itime,nullptr);
        return timerfd;
    }
    int ReadTimerfd(){
        uint64_t times;
        //在处理其它描述符时可能耗时比较多，所以在处理定时器事件时可能已经超时多次
        //read读取定时器描述符的返回值，就是超时的次数
        int ret=read(_timerfd,&times,sizeof(times));
        if(ret<0){
            ERR_LOG("ReadTimerFd error: %s\n",strerror(errno));
            abort();
        }
        return times;
    }
    //这个函数应该每秒执行一次，相当于秒针向后走一步
    void RunTimerTask(){
        _tick=(_tick+1)%_capacity;
        _wheel[_tick].clear();
    }
    //根据超时次数，执行对应的超时任务，实际上我们这个类要关注的任务就是按时让指针向后移动，
    //移动后就销毁轮子所到位置的vector成员，每个成员的析构中对应着一个定时任务，从而触发定时任务
    void Ontime(){
        int times=ReadTimerfd();
        for(int i=0;i<times;i++){
            RunTimerTask();
        }
    }
    void TimerAddInLoop(uint64_t id,uint32_t delay,TaskFunc cb){
        PtrTask pt(new TimerTask(id,delay,cb));
        pt->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,id));
        int tick=(_tick+delay)%_capacity;
        _wheel[tick].push_back(pt);
        _timers[id]=WeakTask(pt);
    }
    void TimerRefreshInLoop(uint64_t id){
        auto it=_timers.find(id);
        if(it==_timers.end()){
            std::cout<<"未找到定时任务，没法刷新，没法延迟";
            return;
        }
        PtrTask pt=it->second.lock();
        int delay=pt->Delaytime();
        int tick=(_tick+delay)%_capacity;
        _wheel[tick].push_back(pt);
    }
    void TimerCancelInLoop(uint64_t id){
        auto it=_timers.find(id);
        if(it==_timers.end()){
            return;
        }
        PtrTask pt=it->second.lock();
        if(pt) pt->Cancel();
    }
    public:
    TimerWheel(EventLoop* loop)
    :_capacity(60),_tick(0),_wheel(_capacity),_loop(loop),_timerfd(CreateTimerfd()),_timer_channel(new Channel(_loop,_timerfd)){
        _timer_channel->SetReadcallback(std::bind(&TimerWheel::Ontime,this));
        _timer_channel->EnableRead();   //启动读事件监控
    }

    //定时器中有个_timers成员，定时器信息的操作可能在多个线程中进行，所以会有线程安全的问题
    //如果不想加锁解决的话，我们就需要把对定期的所有操作都放在同一个线程中解决
    void TimerAdd(uint64_t id,uint32_t delay,TaskFunc cb);
    void TimerRefresh(uint64_t id);
    void TimerCancel(uint64_t id);
    //这几个接口函数存在线程安全问题，所以不能直接被外部调用，需要通过 EventLoop 成员来管理调用它
    bool HashTimer(uint64_t id){
        auto it=_timers.find(id);
        if(it==_timers.end()){
            return false;
        }
        return true;
    }  //这个接口存在线程安全问题，只能在对应的 EventLoop 线程内执行
};



//我们要实现的Reactor模型是One Thread One Loop，这个EventLoop类就是对我们的Pool进行封装
//确保一个Pool监控的事件都是在同一个线程下进行的
class EventLoop{
    private:
    using Functor=std::function<void()>;
    std::thread::id _thread_id;
    int _event_fd;      //唤醒IO事件监控有可能导致的阻塞
    std::unique_ptr<Channel> _event_channel;
    Pool _poller;
    std::mutex _mutex;
    std::vector<Functor> _tasks;
    TimerWheel _timer_wheel;  //定时器模块
    public:
    //执行任务池中的所有任务
    void RunAllTack(){
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            functor.swap(_tasks);
        }
        for(auto& f:functor){
            f();
        }
        return;
    }
    static int CreateEventfd(){
        int fd=eventfd(0,EFD_CLOEXEC | EFD_NONBLOCK);
        if(fd<0){
            ERR_LOG("eventfd error: %s\n",strerror(errno));
            abort();
        }
        return fd;
    }
    void ReadEventfd(){
        uint64_t res=0;
        int ret=read(_event_fd,&res,sizeof(res));
        if(ret<=0){
            if(errno==EINTR | errno==EAGAIN){
                return;
            }
            ERR_LOG("ReadEventfd error: %s\n",strerror(errno));
            abort();
        }
        return;
    }
    void WakeUpEventfd(){
        uint64_t val=1;
        int ret=write(_event_fd,&val,sizeof(val));
        if(ret<0){
            if(errno==EINTR){
                return;
            }
            ERR_LOG("WakeUpEvent error: %s\n",strerror(errno));
            abort();
        }
        return;
    }
    public:
    EventLoop():_thread_id(std::this_thread::get_id()),
                _event_fd(CreateEventfd()),
                _event_channel(new Channel(this,_event_fd)),
                _timer_wheel(this){
                    _event_channel->SetReadcallback(std::bind(&EventLoop::ReadEventfd,this));
                    _event_channel->EnableRead();
                }
    //三步走——1.事件监控 2.就绪事件处理 3.执行任务
    void Start(){
        while(1){
            //1.事件监控
            std::vector<Channel*> active;
            _poller.Poll(&active);
            //2.事件处理
            for(auto& a:active){
                a->HandleEvent();
            }
            //3.执行任务
            RunAllTack();
        }
    }
    bool IsInLoop(){
        return (_thread_id==std::this_thread::get_id());
    }
    void AssertInLoop(){
        assert(_thread_id==std::this_thread::get_id());
    }
    //判断要执行的任务是否在当前线程内，如果是就执行，如果不是则压入任务队列
    void RunInLoop(const Functor& cb){
        if(IsInLoop()){
            return cb();
        }
        return QueueInLoop(cb);
    }
    void QueueInLoop(const Functor& cb){
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(cb);
        }
        WakeUpEventfd();
    }
    void UpdateEvent(Channel* channel){
        return _poller.UpdateEvent(channel);
    }
    void RemoveEvent(Channel* channel){
        return _poller.RemoveEvent(channel);
    }

    //定时器
    void TimerAdd(uint64_t id,uint32_t delay,const TaskFunc& cb){
        return _timer_wheel.TimerAdd(id,delay,cb);
    }
    void TimerRefresh(uint64_t id) { return _timer_wheel.TimerRefresh(id); }
    void TimerCancle(uint64_t id) { return _timer_wheel.TimerCancel(id); }
    bool HashTimer(uint64_t id) { return _timer_wheel.HashTimer(id); }
};
//"Loop-per-Thread": LoopThread封装了"一个线程对应一个EventLoop"的模式，确保每个EventLoop都在自己专属的线程中运行，
//同时提供线程安全的Loop获取机制。
class LoopThread{
    private:
    std::mutex _mutex;
    std::condition_variable _cond;
    EventLoop* _loop;
    std::thread _thread;
    private:
    void ThreadEntry(){
        EventLoop loop;      // 1. 在线程栈上创建EventLoop对象
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop=&loop;
            _cond.notify_all();
        }
        loop.Start();        //  启动事件循环（阻塞在这里）
        // 注意：loop.Start() 返回时，loop对象已析构，_loop成为悬空指针！
    }
    public:
    LoopThread():_loop(nullptr),_thread(&LoopThread::ThreadEntry,this)
    {}
    EventLoop* GetLoop(){
        EventLoop* loop=nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock,[&](){
                return _loop!=nullptr;
            });
            loop=_loop;
        }
        return loop;
    }
};
//线程池
class LoopThreadPool{
    private:
    int _thread_count;
    int _next_idx;
    EventLoop* _baseloop;
    std::vector<LoopThread*> _threads;
    std::vector<EventLoop*> _loops;
    public:
    LoopThreadPool(EventLoop* baseloop)
    :_thread_count(0),_next_idx(0),_baseloop(baseloop)
    {}
    void SetThreadCount(int count) { _thread_count=count; }
    void Create(){
        if(_thread_count>0){
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for(int i=0;i<_thread_count;i++){
                _threads[i]=new LoopThread();
                _loops[i]=_threads[i]->GetLoop();
            }
        }
        return;
    }
    EventLoop* NextLoop(){
        if(_thread_count==0){
            return _baseloop;
        }
        _next_idx=(_next_idx+1)%_thread_count;
        return _loops[_next_idx];
    }
};


class Any{
    private:
    class placeholder{
        public:
        virtual ~placeholder() {}
        virtual const std::type_info& type()=0;
        virtual placeholder* clone()=0;
    };
    template<class T>
    class holder:public placeholder{
        public:
        T _data;
        public:
        holder(const T& data):_data(data) {}
        ~holder() {}
        const std::type_info& type(){
            return typeid(T);
        }
        placeholder* clone(){
            return new holder(_data);
        }
    };
    placeholder* _content;
    public:
    Any() :_content(nullptr){}
    template<class T>
    Any(const T& val):_content(new holder<T>(val)) {}
    Any(const Any& other){
        _content=other._content->clone();
    }
    ~Any(){
        if(_content) delete _content;
    }
    const std::type_info& type(){
        return (_content!=nullptr)?_content->type():typeid(void);
    }
    Any& swap(Any& other){
        std::swap(_content,other._content);
        return *this;
    }

    template<class T>
    Any operator=(const T& val){
        Any(val).swap(*this);
        return *this;
    }
    Any operator=(Any other){
        other.swap(*this);
        return *this;
    }
    template<class T>
    T* get(){
        assert(typeid(T)==_content->type());
        return &((holder<T>*)_content)->_data;
    }
};
// Connection类代表一个TCP连接，它使用事件驱动模型（基于EventLoop和Channel）来处理网络事件（读、写、关闭、错误等）。它提供了以下功能：
// 1. 管理连接的生命周期（建立、关闭、状态转换）。
// 2. 数据的接收和发送（使用输入输出缓冲区）。
// 3. 设置回调函数来处理连接建立、消息到达、连接关闭等事件。
// 4. 支持非活跃连接自动销毁（通过定时器）。
// 5. 支持协议升级（动态更换回调函数和上下文）。
class Connection;
typedef enum {DISCONNECTED,CONNECTING,CONNECTED,DISCONNECTING} ConnStatu;
//DISCONNECTED → CONNECTING → CONNECTED → DISCONNECTING → DISCONNECTED
//DISCONNECTED 连接关闭状态    CONNECTING 连接建立成功-待处理   CONNECTED 连接建立成功，各种设置已完成，可以通信   DISCONNECTING  待关闭
using PtrConnection=std::shared_ptr<Connection>;
class Connection:public std::enable_shared_from_this<Connection>{
    private:
    uint64_t _conn_id;    //连接的唯一标识符，用于管理和查找连接
    //uint64_t _timer_id;   //定时器id，这个是需要我们手动传入且也必须是唯一的，这里我们就可以直接使用连接id充当
    int _sockfd;    //连接对应的套接字文件描述符。
    bool _enable_inactive_release;    //布尔标志，表示是否启用非活跃连接销毁机制，默认为false
    EventLoop* _loop;    //指向EventLoop的指针，该连接所属的事件循环
    ConnStatu _statu;    //连接状态
    Socket _socket;     //Socket对象，用于套接字操作（如关闭、非阻塞IO等）
    Channel _channel;   //Channel对象，用于管理套接字的事件监听和回调
    Buffer _in_buffer;    //输入缓冲区，存放从套接字读取的数据
    Buffer _out_buffer;   //输出缓冲区，存放待发送的数据
    Any _context;    //任意类型的上下文数据，用于存储与连接相关的用户数据
    using ConnectedCallback=std::function<void(const PtrConnection&)>;
    using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;
    using CloseCallback=std::function<void(const PtrConnection&)>;
    using AnyEventCallback=std::function<void(const PtrConnection&)>;
    ConnectedCallback _connected_callback;   //连接建立完成回调
    MessageCallback  _message_callback;    //消息到达回调
    CloseCallback _close_callback;      //连接关闭回调
    AnyEventCallback _event_callback;    //任意事件回调
    CloseCallback _server_close_callback;     //服务器内部关闭回调，用于从服务器连接管理中移除自身
    private:
    //1. 事件处理函数
    //描述符可读事件触发后调用的函数
    void HandleRead(){
        char buffer[65536];
        int ret=_socket.NonBlockRecv(buffer,65535);
        if(ret<0){
            return Shutdown();    //读取错误，准备关闭
        }
        _in_buffer.WriteAndPush(buffer,65535);
        if(_in_buffer.ReadAbleSize()>0){
            std::cout << "HandleRead" << std::endl;
            _message_callback(shared_from_this(),&_in_buffer);
        }
    }
    void HandleWrite(){
        //_out_buffer中保存的数据就是要发送的数据
        int ret=_socket.NonBlockSend(_out_buffer.ReadPosition(),_out_buffer.ReadAbleSize());
        if(ret<0){
            //发送错误，处理剩余数据后关闭
            if(_in_buffer.ReadAbleSize()>0){
                _message_callback(shared_from_this(),&_in_buffer);
            }
            return Release();
        }
        _out_buffer.MoveReadOffset(ret);
        if(_out_buffer.ReadAbleSize()==0){
            _channel.DisableWrite();    //没有数据发送了，关闭写事件监控（避免频繁的写事件监控，只有在有数据的时候监控）
            if(_statu==DISCONNECTING){   //如果正在关闭，立即释放
                return Release(); 
            }
        }
        return;
    }
    void HandleClose(){
        if(_in_buffer.ReadAbleSize()>0){
            _message_callback(shared_from_this(),&_in_buffer);
        }
        return Release();
    }
    void HandleError(){
        return HandleClose();
    }
    void HandleEvent(){
        if(_enable_inactive_release==true){
            _loop->TimerRefresh(_conn_id);
        }
        if(_event_callback) { _event_callback(shared_from_this()); }
    }
    //2. 连接生命周期管理
    //完成连接建立的最后步骤，启动读监控
    void EstablishedInLoop(){
        assert(_statu==CONNECTING);
        _statu=CONNECTED;
        _channel.EnableRead();
        if(_connected_callback) { _connected_callback(shared_from_this()); }
    }
    void ReleaseInLoop(){
        _statu=DISCONNECTED;    //修改状态
        _channel.Remove();     //移除监控
        _socket.Close();     //关闭socket
        //清理定时器
        if(_loop->HashTimer(_conn_id)) { CancelInactiveReleaseInLoop(); }
        //回调通知
        if(_close_callback) { _close_callback(shared_from_this()); }
        if(_server_close_callback) { _server_close_callback(shared_from_this()); }
    }
    void ShutdownInLoop(){
        _statu=DISCONNECTING;
        if(_in_buffer.ReadAbleSize()>0){
            _message_callback(shared_from_this(),&_in_buffer);
        }
        if(_out_buffer.ReadAbleSize()>0){
            if(_channel.WriteAble()==false){
                _channel.EnableWrite();
            }
        }
        if(_out_buffer.ReadAbleSize()==0){
            return Release();
        }
    }
    //3. 发送数据
    void SendInLoop(Buffer& buffer){
        if(_statu==DISCONNECTED) return;
        _out_buffer.WriteBufferAndPush(buffer);
        if(_channel.WriteAble()==false){
            _channel.EnableWrite();
        }
    }
    //4. 非活跃连接管理
    void EnableInactiveReleaseInLoop(int sec){
        _enable_inactive_release=true;
        if(_loop->HashTimer(_conn_id)){
            return _loop->TimerRefresh(_conn_id);
        }
        _loop->TimerAdd(_conn_id,sec,std::bind(&Connection::Release,this));
    }
    void CancelInactiveReleaseInLoop(){
        _enable_inactive_release=false;
        if(_loop->HashTimer(_conn_id)){
            _loop->TimerCancle(_conn_id);
        }
    }
    //5. 协议升级
    void UpgradeInLoop(const Any& context,const ConnectedCallback& con,
    const MessageCallback& msg,const CloseCallback& close,const AnyEventCallback& event){
        _context=context;
        _connected_callback=con;
        _message_callback=msg;
        _close_callback=close;
        _event_callback=event;
    }
    public:
    Connection(EventLoop* loop,int conn_id,int sockfd)
    :_conn_id(conn_id),_sockfd(sockfd),_enable_inactive_release(false),
    _loop(loop),_statu(CONNECTING),_socket(_sockfd),_channel(_loop,_sockfd){
        _channel.SetReadcallback(std::bind(&Connection::HandleRead,this));
        _channel.SetWritecallback(std::bind(&Connection::HandleWrite,this));
        _channel.SetClosecallback(std::bind(&Connection::HandleClose,this));
        _channel.SetErrorcallback(std::bind(&Connection::HandleError,this));
        _channel.SetEventcallback(std::bind(&Connection::HandleEvent,this));
    }
    ~Connection(){ DEB_LOG("Release connection: %p",this); }
    int Fd(){ return _sockfd; }
    int Id(){ return _conn_id; }
    bool Connected(){ return _statu; }
    void SetContext(const Any& context){ _context=context; }
    Any* GetContext(){ return &_context; }
    void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
    void SetClosedCallback(const CloseCallback&cb) { _close_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
    void SetsvrCloseCallback(const CloseCallback&cb) { _server_close_callback = cb; }
    void Established(){
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
    }
    void Send(const char* data,size_t len){
        Buffer buffer;
        buffer.WriteAndPush(data,len);
        _loop->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buffer)));
    }
    void Shutdown(){
        _loop->RunInLoop(std::bind(&Connection::ShutdownInLoop,this));
    }
    void Release(){
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop,this));
    }
    void EnableInactiveRelease(int sec){
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
    }
    void CancleInactiveRelease(){
        _loop->RunInLoop(std::bind(&Connection::CancelInactiveReleaseInLoop,this));
    }
    //切换协议，这个接口必须在EventLoop中执行，防止新的事件触发处理的时候，切换任务还没有被执行，导致这部分数据按照原协议处理
    void Upgrade(const Any& context,const ConnectedCallback& con,
    const MessageCallback& msg,const CloseCallback& close,const AnyEventCallback& event){
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,con,msg,close,event));
    }
};


//接收器类——进行监听描述符管理
class Acceptor{
    private:
    Socket _socket;     //⽤于创建监听套接字
    EventLoop* _loop;     //⽤于对监听套接字进⾏事件监控
    Channel _channel;    //⽤于对监听套接字进⾏事件管理
    
    using AcceptCallback=std::function<void(int)>;
    AcceptCallback _accept_callback;   //新连接接收回调函数
    private:
    void HandleRead(){
        int newfd=_socket.Accept();
        if(newfd<0){
            return;
        }
        if(_accept_callback) { _accept_callback(newfd); }
    }
    int CreateServer(int port){
        bool ret=_socket.CreateServer(port);
        assert(ret==true);
        return _socket.Fd();
    }
    public:
    Acceptor(EventLoop* loop,int port)
    :_loop(loop),_socket(CreateServer(port)),_channel(loop,_socket.Fd()){
        _channel.SetReadcallback(std::bind(&Acceptor::HandleRead,this));
    }
    void SetAcceptCallbakc(const AcceptCallback& cb) { _accept_callback=cb; }
    void Listen() { _channel.EnableRead(); }  
    //启动读事件监控不能在初始化中就进行，防止我们一启动，就有读事件到来，此时我们的回调函数_accept_callback如果还没有传入赋值而出错
    // 使用时的正确顺序：
    // Acceptor acceptor(loop, 8080);        // 1. 创建Acceptor
    // acceptor.SetAcceptCallback(callback);  // 2. 设置回调函数  
    // acceptor.Listen();                     // 3. 启动监控
};


void Channel::Remove() { return _loop->RemoveEvent(this); }
void Channel::Update() { return _loop->UpdateEvent(this); }
//添加、刷新、延迟定时任务
void TimerWheel::TimerAdd(uint64_t id,uint32_t delay,TaskFunc cb){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,id,delay,cb));
}
void TimerWheel::TimerRefresh(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,id));
}
void TimerWheel::TimerCancel(uint64_t id){
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,id));
}

//TCP服务器类设计
class TcpServer{
    private:
    //连接管理相关: 全局管理所有活跃连接，便于查找、统计和清理。
    uint64_t _next_id;     // 自增连接ID，确保每个连接唯一标识
    std::unordered_map<uint64_t,PtrConnection> _conns;      // 所有连接的管理容器
    //服务器配置相关: 配置服务器行为和连接保活策略。
    int _port;    //监听端口
    int _timeout;    //非活跃连接超时时间
    bool _enable_inactive_release;    //是否启用连接超时释放
    //核心组件相关：构成服务器的事件驱动架构基础。
    EventLoop _baseloop;     //主EventLoop，运行在主线程
    Acceptor _acceptor;     //接受器，处理新连接请求
    LoopThreadPool _pool;     //从属EventLoop线程池

    using ConnectedCallback=std::function<void(const PtrConnection&)>;
    using MessageCallback=std::function<void(const PtrConnection&,Buffer*)>;
    using CloseCallback=std::function<void(const PtrConnection&)>;
    using AnyEventCallback=std::function<void(const PtrConnection&)>;
    using Functor=std::function<void()>;
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    CloseCallback _closed_callback;
    AnyEventCallback _event_callback;
    private:
    //新连接处理 - 核心分发逻辑
    void NewConnection(int fd){
        _next_id++;
        PtrConnection conn(new Connection((_pool.NextLoop()),_next_id,fd));
        //这里必须动态开辟，因为我们需要传指针给回调函数，且 HandleClose 还要对这个对象进行销毁，如果不动态开辟空间
        //直接创建一个对象，那么它就存在在栈空间上，不能够使用 delete 销毁
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetConnectedCallback(_connected_callback);
        conn->SetAnyEventCallback(_event_callback);
        conn->SetsvrCloseCallback(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
        //非活跃连接的超时释放操作，10秒后关闭连接
        //注意定时销毁任务，必须在启动读事件监控之前，以防止我们读事件监控刚启动就有读事件，而我们的定时销毁任务还没有启动而没有刷新定时销毁任务
        if(_enable_inactive_release) conn->EnableInactiveRelease(_timeout);
        conn->Established();
        _conns.insert(std::make_pair(_next_id,conn));
    }
    void RemoveConnectionInLoop(const PtrConnection& conn){
        int id=conn->Id();
        auto it=_conns.find(id);
        if(it!=_conns.end()){
            _conns.erase(it);
        }
    }
    void RemoveConnection(const PtrConnection& conn){
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
    }

    //定时任务
    void RunAfterInLoop(const Functor& task,int delay){
        _next_id++;
        _baseloop.TimerAdd(_next_id,delay,task);
    }
    public:
    //一次性完成所有组件的初始化与关联
    TcpServer(int port)
    :_port(port),_next_id(0),_enable_inactive_release(false),_acceptor(&_baseloop,port),_pool(&_baseloop){
        _acceptor.SetAcceptCallbakc(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
        _acceptor.Listen();
    }
    void SetThreadCount(int count) { return _pool.SetThreadCount(count); }
    void SetConnectedCallback(const ConnectedCallback&cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback&cb) { _message_callback = cb; }
    void SetClosedCallback(const CloseCallback&cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback&cb) { _event_callback = cb; }
    
    void EnableInactiveRelease(int timeout){
        _timeout=timeout;
        _enable_inactive_release=true;
    }
    void RunAfter(const Functor& task,int delay){
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
    }
    void Start(){
        _pool.Create();
        _baseloop.Start();
    }
};
